package com.bff.gaia.streaming.api.xjoin.client.hbase.core;

import com.bff.gaia.streaming.api.xjoin.client.hbase.core.enums.EReplaceType;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Lists;

import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class RowKeyBuilder implements Serializable {

	private static Pattern Md5Operator = Pattern.compile("(?i)^md5\\(\\s*(.*)\\s*\\)$");

	private List<List<ReplaceInfo>> operatorChain = Lists.newArrayList();

	public void init(String rowKeyTempl){

		String[] strArr = splitIgnoreQuotaBrackets(rowKeyTempl, "\\+");

		for(String infoAlias : strArr){
			infoAlias = infoAlias.trim();
			Matcher matcher = Md5Operator.matcher(infoAlias);
			String fieldCols = null;
			if(matcher.find()){
				fieldCols = matcher.group(1);
			}else{
				fieldCols = infoAlias;
			}

			String[] fieldArr = fieldCols.split("\\+");
			List<ReplaceInfo> fieldList = Lists.newArrayList();
			for(String oneField : fieldArr){
				ReplaceInfo replaceInfo =  getReplaceInfo(oneField);
				fieldList.add(replaceInfo);
			}

			operatorChain.add(fieldList);
		}
	}

	public String getRowKey(Map<String, Object> refData){

		StringBuilder sb = new StringBuilder("");
		for(List<ReplaceInfo> fieldList : operatorChain){
			sb.append(buildStr(fieldList, refData));
		}

		return sb.toString();
	}

	private String buildStr(List<ReplaceInfo> fieldList, Map<String, Object> refData){
		StringBuffer sb = new StringBuffer("");
		for(ReplaceInfo replaceInfo : fieldList){

			if(replaceInfo.getType() == EReplaceType.CONSTANT){
				sb.append(replaceInfo.getParam());
				continue;
			}

			String replaceName = replaceInfo.getParam();
			if(!refData.containsKey(replaceName)){
				throw new RuntimeException(String.format("build rowKey with field %s which value not found.", replaceName));
			}

			sb.append(refData.get(replaceName));
		}

		return sb.toString();
	}

	public static String[] splitIgnoreQuotaBrackets(String str, String delimiter){
		String splitPatternStr = delimiter + "(?![^()]*+\\))(?![^{}]*+})(?![^\\[\\]]*+\\])";
		return str.split(splitPatternStr);
	}

	public ReplaceInfo getReplaceInfo(String field){

		field = field.trim();
		if(field.length() <= 2){
			throw new RuntimeException(field + " \n" +
				"Format defined exceptions");
		}

		//判断是不是常量==>''包裹的标识
		if(field.startsWith("'") && field.endsWith("'")){
			ReplaceInfo replaceInfo = new ReplaceInfo(EReplaceType.CONSTANT);
			field = field.substring(1, field.length() - 1);
			replaceInfo.setParam(field);
			return replaceInfo;
		}

		ReplaceInfo replaceInfo = new ReplaceInfo(EReplaceType.PARAM);
		replaceInfo.setParam(field);
		return replaceInfo;
	}
}